package mr

import "fmt"
import "log"
import "net"
import "net/rpc"
import "net/http"
import "hash/fnv"
import "os"
import "time"
import "strconv"
import "io/ioutil"
import "encoding/json"
import "sort"

//
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
    Key   string
    Value string
}

type Ping struct {
    ph bool
}

func (p *Ping) ping() {
    rpc.Register(p)
    rpc.HandleHTTP()
    sockname := cookupSock(os.Getpid())
    os.Remove(sockname)
    l, e := net.Listen("unix", sockname)
    if e != nil {
        log.Fatal("listen error:", e)
    }
    go http.Serve(l, nil)
}

func (p *Ping) Tic(args *bool, reply *bool) error {
    (*reply) = true
    return nil
}

type ByKey []KeyValue
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

func ihash(key string) int {
    h := fnv.New32a()
    h.Write([]byte(key))
    return int(h.Sum32() & 0x7fffffff)
}

func LOGCF() {
    log.Fatalf("Failed Contacting Master")
}

//
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
    reducef func(string, []string) string) {
    p_agent := Ping{}
    p_agent.ping()
    nMap, nReduce := call_get_basics()
    // Your worker implementation here.
    pid := os.Getpid()
    for ; true; {
        job_id, file_name, job_type := call_assign_job(pid)
        for ; job_type == -1; {
            time.Sleep(10 * time.Millisecond)
            job_id, file_name, job_type = call_assign_job(pid)
        }
        if job_type == 0 {
            do_map(mapf, job_id, file_name, nMap, nReduce)
        } 
        if job_type == 1 {
            do_reduce(reducef, job_id, nMap, nReduce)
        }
    }
}

func call_get_basics() (int, int) {
    args := 0
    reply := PairReply{}
    if call("Coordinator.GetBasics", &args, &reply) {
        return reply.A, reply.B
    } else {
        LOGCF()
        return -1, -1
    }
}

func call_assign_job(pid int) (int, string, int) {
    args := AssignJobArgs{pid}
    reply := AssignJobReply{}
    if call("Coordinator.AssignJob", &args, &reply) {
        return reply.Id, reply.Name, reply.Job
    } else {
        LOGCF()
        return -1, "ph", -1
    }
}

func call_map_commit(job_id int) (bool) {
    var reply bool
    if call("Coordinator.MapCommit", &job_id, &reply) {
        return reply
    } else {
        LOGCF()
        return false
    }
}

func call_reduce_commit(job_id int) (bool) {
    var reply bool
    if call("Coordinator.ReduceCommit", &job_id, &reply) {
        return reply
    } else {
        LOGCF()
        return false
    }
}

func open_med_files(pid int, nReduce int) ([]*os.File) {
    ret := []*os.File{}
    s_pid := strconv.Itoa(pid)
    for i := 0; i < nReduce; i++ {
        s_slice := strconv.Itoa(i)
        file, err := os.OpenFile("mr_"+s_slice+"_"+s_pid, os.O_WRONLY | os.O_CREATE | os.O_TRUNC, 0777)
        if err != nil {
            fmt.Println("no1")
            log.Println(err)
        }
        ret = append(ret, file)
    }
    return ret
}

func rename_all_files(pid, job_id, nReduce int) {
    s_pid := strconv.Itoa(pid)
    s_job_id := strconv.Itoa(job_id)
    for i := 0; i < nReduce; i++ {
        s_slice := strconv.Itoa(i)
        err := os.Rename("mr_"+s_slice+"_"+s_pid, "mr_"+s_slice+"_"+s_job_id)
        if (err != nil) {
            fmt.Println("no2")
            fmt.Println(err)
        }
    }
}

func do_map(mapf func(string, string) []KeyValue,
            job_id int,
            file_name string,
            nMap int,
            nReduce int) {
    pid := os.Getpid()
    file, err := os.Open(file_name)
    if err != nil {
        log.Fatalf("cannot open %v", file_name)
    }
    content, err := ioutil.ReadAll(file)
    if err != nil {
        log.Fatalf("cannot read %v", file_name)
    }
    file.Close()
    kvlist := mapf(file_name, string(content))
    file_ptr_list := open_med_files(pid, nReduce)
    encs := []*json.Encoder{}
    for i := 0; i < nReduce; i++ {
        encs = append(encs, json.NewEncoder(file_ptr_list[i]))
    }
    L := len(kvlist)
    for i := 0; i < L; i++ {
        r_id := ihash(kvlist[i].Key) % nReduce
        err := encs[r_id].Encode(&kvlist[i])
        if err != nil {
            fmt.Println(err)
        }
    }
    for i := 0; i < nReduce; i++ {
        file_ptr_list[i].Close()
    }
    rename_all_files(pid, job_id, nReduce)
    res := call_map_commit(job_id)
    if res {

    }
}

func do_reduce(reducef func(string, []string) string,
               job_id int,
               nMap int,
               nReduce int) {
    pid := os.Getpid()
    s_pid := strconv.Itoa(pid)
    s_job_id := strconv.Itoa(job_id)
    MED := []KeyValue{}
    for i := 0; i < nMap; i++ {
        s_slice := strconv.Itoa(i)
        file, err := os.Open("mr_"+s_job_id+"_"+s_slice)
        if (err != nil) {
            log.Fatalf("190: %v\n", err)
        }
        var kv KeyValue
        dec := json.NewDecoder(file)
        for {
            if err := dec.Decode(&kv); err != nil {
                break
            }
            MED = append(MED, kv)
        }
    }
    sort.Sort(ByKey(MED))
    oname := "mr-out-"+s_pid
    ofile, _ := os.Create(oname)
    L := len(MED)
    for i := 0; i < L;  {
        j := i + 1
        for j < L && MED[j].Key == MED[i].Key {
            j++
        }
        values := []string{}
        for k := i; k < j; k++ {
            values = append(values, MED[k].Value)
        }
        output := reducef(MED[i].Key, values)
        fmt.Fprintf(ofile, "%v %v\n", MED[i].Key, output)
        i = j
    }
    err := os.Rename("mr-out-"+s_pid, "mr-out-"+s_job_id)
    if err != nil {
        fmt.Println(err)
    }
    res := call_reduce_commit(job_id)
    if res {
    }
}

//
// example function to show how to make an RPC call to the coordinator.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

    // declare an argument structure.
    args := ExampleArgs{}

    // fill in the argument(s).
    args.X = 99

    // declare a reply structure.
    reply := ExampleReply{}

    // send the RPC request, wait for the reply.
    call("Coordinator.Example", &args, &reply)

    // reply.Y should be 100.
    fmt.Printf("reply.Y %v\n", reply.Y)
}

func call(rpcname string, args interface{}, reply interface{}) bool {
    // c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
    sockname := coordinatorSock()
    c, err := rpc.DialHTTP("unix", sockname)
    if err != nil {
        log.Fatal("dialing:", err)
    }
    defer c.Close()

    err = c.Call(rpcname, args, reply)
    if err == nil {
        return true
    }

    fmt.Println(err)
    return false
}
